package org.example.consumer;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class KafkaConsumer {

    @KafkaListener(topics = "springboot", groupId = "default1")
    public void listen(String msg) {
        log.info(msg);
    }


    @KafkaListener(topics = "springboot2", groupId = "default1")
    public void listenGroup2(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        log.info(value);
        // 手动提交offset
        ack.acknowledge();
    }


    @KafkaListener(groupId = "default1", topicPartitions = {
        @TopicPartition(topic = "springboot3", partitions = {"0"}),
        @TopicPartition(topic = "springboot4", partitions = {"0"}, partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
    })
    public void listenGroup3(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        log.info(value);
        // 手动提交offset
        ack.acknowledge();
    }



}
